package com.intct.flink;

import com.intct.hbase.bean.Student;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author gufg
 * @since 2025-06-25 15:52
 */
public class AggregationDemo {
    public static void main(String[] args) throws Exception {
        // 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env.setRuntimeMode(RuntimeExecutionMode.BATCH);

        // 数据源
        DataStreamSource<Student> source = env.fromElements(
                new Student("c_01", "s_001", 3),
                new Student("c_02", "s_002", 1),
                new Student("c_03", "s_003", 1),
                new Student("c_01", "s_004", 1),
                new Student("c_01", "s_005", 2)
        );

        // 分组(分区)
        KeyedStream<Student, String> keyByDS = source.keyBy(new KeySelector<Student, String>() {
            @Override
            public String getKey(Student value) throws Exception {
                return value.getClassId();
            }
        });

        // 聚合 -- sum
//        SingleOutputStreamOperator<Studenet> sumDS = keyByDS.sum("score");
//        sumDS.print();

        /**
         * min 和minBy区别
         *  min : 除了分区字段 和 聚合字段,其他所有字段取第一条数据值
         *  miBy : 取最小聚合字段所在的记录所有字段
         */
        // 聚合 -- min
        SingleOutputStreamOperator<Student> minDS = keyByDS.min("score");
        minDS.print("Min   >>>>>>>");

        // 聚合 -- minBy
        SingleOutputStreamOperator<Student> minByDS = keyByDS.minBy("score");
        minByDS.print("MinBy >>>>>>>");


        // 启动作业
        env.execute();
    }
}
